package com.zidongxiangxi.rabbit.starter.configuration.producer;

import com.zidongxiangxi.rabbit.producer.config.ProducerResendConfig;
import com.zidongxiangxi.rabbit.producer.config.ProducerSequenceClearConfig;
import com.zidongxiangxi.rabbit.producer.job.ProducerResendThread;
import com.zidongxiangxi.rabbit.producer.job.ProducerSequenceClearThread;
import com.zidongxiangxi.rabbit.producer.manager.ProducerManager;
import com.zidongxiangxi.rabbit.producer.manager.ProducerSequenceManager;
import com.zidongxiangxi.rabbit.producer.manager.RabbitTemplateManager;
import com.zidongxiangxi.rabbit.producer.support.RabbitSequenceMessagePostProcessor;
import com.zidongxiangxi.rabbit.producer.support.RabbitTemplateBeanPostProcessor;
import com.zidongxiangxi.rabbit.producer.transaction.listener.RabbitTransactionListener;
import com.zidongxiangxi.rabbit.producer.transaction.listener.TransactionListener;
import com.zidongxiangxi.rabbit.starter.entity.ReliableRabbitProperties;
import com.zidongxiangxi.rabbit.starter.entity.producer.ReliableRabbitProducerRely;
import com.zidongxiangxi.rabbit.starter.entity.producer.ReliableRabbitProducerSequence;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;

import java.util.List;

/**
 * RabbitMq发送配置
 *
 * @author chenxudong
 * @date 2020/11/08
 */
@Configuration
@ConditionalOnProperty(prefix = "reliable-rabbit.producer.rely", name = "enabled", havingValue = "true")
public class ReliableRabbitProduceRelyConfiguration {
	private final ReliableRabbitProperties properties;

	public ReliableRabbitProduceRelyConfiguration(ReliableRabbitProperties properties) {
		this.properties = properties;
	}

	/**
	 * 定义mq消息的manager
	 *
	 * @param jdbcTemplate jdbcTemplate实例
	 * @return mq消息的数据库manager
	 */
	@Bean
	@ConditionalOnMissingBean(ProducerManager.class)
	public ProducerManager rabbitProducerManager(JdbcTemplate jdbcTemplate) {
		return new ProducerManager(properties.getProducer().getRely().getTableName(), jdbcTemplate);
	}

	/**
	 * 定义rabbitTemplate管理类
	 *
	 * @return rabbitTemplate管理类
	 */
	@Bean
	@ConditionalOnMissingBean(name = "rabbitTemplateManager")
	public RabbitTemplateManager rabbitTemplateManager() {
		return new RabbitTemplateManager();
	}

	/**
	 * 定义重发信息的工作线程
	 *
	 * @param producerManager mq消息的manager
	 * @param rabbitSequenceMessagePostProcessorObjectProvider 顺序消息加工处理提供者
	 * @return 重发信息的工作线程
	 */
	@Bean
	@ConditionalOnMissingBean(ProducerResendThread.class)
	public ProducerResendThread producerResendThread(
			ProducerManager producerManager,
			ObjectProvider<RabbitSequenceMessagePostProcessor> rabbitSequenceMessagePostProcessorObjectProvider
	) {
		ReliableRabbitProducerRely relyProperties = properties.getProducer().getRely();
		ProducerResendConfig config = new ProducerResendConfig();
		config.setApplication(properties.getApplication());
		config.setBatchSize(relyProperties.getBatchSize());

		return new ProducerResendThread(config, producerManager, rabbitTemplateManager(),
				rabbitSequenceMessagePostProcessorObjectProvider.getIfAvailable());
	}

	/**
	 * 定义默认的事务监听
	 *
	 * @param producerManager mq消息的manager
	 * @return 默认的事务监听
	 */
	@Bean
	public TransactionListener defaultTransactionListener(
			ProducerManager producerManager,
			ObjectProvider<ProducerSequenceManager> producerSequenceManagerObjectProvider,
			ObjectProvider<RabbitSequenceMessagePostProcessor> rabbitSequenceMessagePostProcessorObjectProvider
	) {
		ProducerSequenceManager producerSequenceManager = producerSequenceManagerObjectProvider.getIfUnique();
		RabbitSequenceMessagePostProcessor rabbitSequenceMessagePostProcessor =
				rabbitSequenceMessagePostProcessorObjectProvider.getIfUnique();
		return new RabbitTransactionListener(producerManager, producerSequenceManager, rabbitSequenceMessagePostProcessor);
	}

	/**
	 * 定义rabbitTemplate后置加工
	 *
	 * @param producerManager 消息生产manager
	 * @return rabbitTemplate后置处理
	 */
	@Bean
	public RabbitTemplateBeanPostProcessor rabbitTemplateBeanPostProcessor(
			ProducerManager producerManager,
			List<TransactionListener> transactionListeners
	) {
		return new RabbitTemplateBeanPostProcessor(properties.getApplication(), producerManager, rabbitTemplateManager(),
				transactionListeners);
	}

	/**
	 * 顺序消息配置
	 */
	@ConditionalOnProperty(prefix = "reliable-rabbit.producer.rely.sequence", name = "enabled", havingValue = "true")
	protected static class ReliableRabbitProducerSequenceConfiguration {
		private final ReliableRabbitProperties properties;

		public ReliableRabbitProducerSequenceConfiguration(ReliableRabbitProperties properties) {
			this.properties = properties;
		}

		/**
		 * 定义顺序消息管理类
		 *
		 * @param jdbcTemplate jdbc模板
		 * @return 顺序消息管理类
		 */
		@Bean
		public ProducerSequenceManager producerSequenceManager(JdbcTemplate jdbcTemplate) {
			return new ProducerSequenceManager(properties.getProducer().getRely().getSequence().getTableName(), jdbcTemplate);
		}

		/**
		 * 定义顺序消息加工处理
		 *
		 * @param producerSequenceManager 顺序消息管理类
		 * @return 顺序消息加工处理
		 */
		@Bean
		public RabbitSequenceMessagePostProcessor rabbitSequenceMessagePostProcessor(ProducerSequenceManager producerSequenceManager) {
			return new RabbitSequenceMessagePostProcessor(producerSequenceManager);
		}

		/**
		 * 定义清理顺序消息记录的工作线程
		 *
		 * @param producerSequenceManager 顺序消息manager
		 * @return 清理顺序消息记录的工作线程
		 */
		@Bean
		public ProducerSequenceClearThread producerSequenceClearThread(ProducerSequenceManager producerSequenceManager) {
			ReliableRabbitProducerSequence sequence = properties.getProducer().getRely().getSequence();
			ProducerSequenceClearConfig config = new ProducerSequenceClearConfig();
			config.setRetentionPeriod(sequence.getRetentionPeriod());
			config.setBatchSize(sequence.getBatchSize());
			config.setWorkPeriod(sequence.getWorkPeriod());

			return new ProducerSequenceClearThread(config, producerSequenceManager);
		}
	}

}
